
#include "stdafx.h"
#include "ConstDef.h"
#include "ReloadCode.h"
#include "ReloadApp.h"
#include "ReloadPacket.h"
#include "ReloadOverlay.h"
#include "ReloadConfigure.h"
#include "ReloadNetworkDataHandler.h"
#include "ReloadConnectionActor.h"
#include "ReloadTopologyPluginInterface.h"
#include "ReloadChord.h"

namespace reload
{
	IOErrorTransactionStatusParam::IOErrorTransactionStatusParam()
	{

	}

	IOErrorTransactionStatusParam::~IOErrorTransactionStatusParam()
	{

	}

	///////////////////////////////////////////////////////////////////////////////////////////////

	ConnectionTable::ConnectionTable()
	:connections_(),nodeIdIndexes_()
	{

	}

	ConnectionTable::~ConnectionTable()
	{
		connections_.clear();
		nodeIdIndexes_.clear();
	}

	retT ConnectionTable::add( boost::weak_ptr<ReloadConnectionActor> _pReloadConnectionActor )
	{
		if(boost::shared_ptr<ReloadConnectionActor> pReloadConnectionActor = _pReloadConnectionActor.lock())
		{
			connections_[pReloadConnectionActor->getOpaqueId()] = _pReloadConnectionActor;
			if(!pReloadConnectionActor->getNodeId().isNULL())
				nodeIdIndexes_[pReloadConnectionActor->getNodeId()] = _pReloadConnectionActor;
		}
		RETSUCCESS;
	}

	///////////////////////////////////////////////////////////////////////////////////////////////

	RoutingTable::RoutingTable( boost::weak_ptr<ReloadOverlay> _pReloadOverlay )
	:pReloadOverlay_(_pReloadOverlay)
	{

	}

	RoutingTable::~RoutingTable()
	{

	}

	////////////////////////////////////////////////////////////////////////////////////////////

	TransactionStage::TransactionStage()
	:nextStage_(TransactionStage_Finish),param_()
	{

	}

	TransactionStage::~TransactionStage()
	{
		nextStage_ = TransactionStage_Invalid;
		param_.reset();
	}

	/////////////////////////////////////////////////////////////////////////////////////////////

	Transaction::Transaction()
	:pReloadNetworkDataContext_(),pTransactionStage_(new TransactionStage()),transactionOperation_(NULL),
	pTimerContext_()
	{

	}

	Transaction::~Transaction()
	{
		pReloadNetworkDataContext_.reset();
		pTransactionStage_.reset();
		transactionOperation_ = NULL;
		pTimerContext_.reset();
	}

	/////////////////////////////////////////////////////////////////////////////////////////////

	TransactionMgr::TransactionMgr( boost::weak_ptr<ReloadOverlay> _pReloadOverlay )
	:pReloadOverlay_(_pReloadOverlay),transactionMap_()
	{

	}

	TransactionMgr::~TransactionMgr()
	{
		pReloadOverlay_.reset();
		transactionMap_.clear();
	}

	TransactionIdType TransactionMgr::generateTransactionId()
	{
		TransactionIdType transactionId;
		do 
		{
			transactionId = generateRandom64(false);
		} while (transactionMap_.find(transactionId) != transactionMap_.end());
		return transactionId;
	}

	TransactionIdType TransactionMgr::beginTransaction( boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext, int _nextStage, boost::shared_ptr<asyframe::WildcardElement> _param )
	{
		TransactionIdType transactionId = generateTransactionId();
		transactionMap_[transactionId].pReloadNetworkDataContext_ = _pReloadNetworkDataContext;
		transactionMap_[transactionId].pTransactionStage_->nextStage_ = _nextStage;
		transactionMap_[transactionId].pTransactionStage_->param_ = _param;
		return transactionId;
	}

	retT TransactionMgr::switchToNextStage( TransactionIdType _transactionId, boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< TransactionIdType, Transaction >::iterator it = transactionMap_.find(_transactionId);
		if(it == transactionMap_.end())
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONID_NOT_EXIST);
		if(!it->second.pReloadNetworkDataContext_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(it->second.transactionStatus_.status_ == TransactionStatus_Operating)
			R(onWaitOperationTimeout(_transactionId, it->second.pTimerContext_, boost::system::error_code(), 0));
		if(!_pReloadNetworkDataContext)
		{
			_pReloadNetworkDataContext.reset(new ReloadNetworkDataContext());
			_pReloadNetworkDataContext->pActor_ = it->second.pReloadNetworkDataContext_->pActor_;
		}
		R(processTransaction(it->second.pTransactionStage_, it->second.transactionStatus_, pReloadOverlay, _pReloadNetworkDataContext));
		transactionMap_.erase_return_void(it);
		RETSUCCESS;
	}

	retT TransactionMgr::processTransaction( boost::shared_ptr<TransactionStage> _pTransactionStage, TransactionStatus _transactionStatus, boost::shared_ptr<ReloadOverlay> _pReloadOverlay, boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		if(!_pReloadOverlay || !_pTransactionStage)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		switch(_pTransactionStage->nextStage_)
		{
		case TransactionStage_Finish:break;
		case TransactionStage_SendAttachReqAfterConnectToBootstrap:_pReloadOverlay->sendAttachReqAfterConnectToBootstrap(_transactionStatus, _pReloadNetworkDataContext);break;
		case TransactionStage_ConnectAfterAttach:_pReloadOverlay->connectAfterAttach(_transactionStatus, _pReloadNetworkDataContext, _pTransactionStage->param_);break;
		case TransactionStage_AddNodeIdToRoutingTableAfterConnect:_pReloadOverlay->addNodeIdToRoutingTableAfterConnect(_transactionStatus, _pReloadNetworkDataContext, _pTransactionStage->param_);break;
		case TransactionStage_PrepareRoutingTableAfterConnectToAdmittingPeer:_pReloadOverlay->prepareRoutingTableAfterConnectToAdmittingPeer(_transactionStatus, _pReloadNetworkDataContext);break;
		case TransactionStage_SendUpdateReqAfterConnect:_pReloadOverlay->sendUpdateReqAfterConnect(_transactionStatus, _pReloadNetworkDataContext);break;
		case TransactionStage_InvokeUserCallBack:_pReloadOverlay->invokeUserCallBack(_transactionStatus, _pReloadNetworkDataContext, _pTransactionStage->param_);break;
		case TransactionStage_SendJoinReqAfterConnectToAdmittingPeer:_pReloadOverlay->sendJoinReqAfterConnectToAdmittingPeer(_transactionStatus, _pReloadNetworkDataContext);break;
		case TransactionStage_RefreshAdmittingPeerAfterConnectToAdmittingPeer:_pReloadOverlay->refreshAdmittingPeerAfterConnectToAdmittingPeer(_transactionStatus, _pReloadNetworkDataContext);break;
		case TransactionStage_SendPingReqAfterConnect:_pReloadOverlay->sendPingReqAfterConnect(_transactionStatus, _pReloadNetworkDataContext);break;
		default:break;
		}
		R(_pReloadOverlay->pTopologyPlugin_->processTransaction(_pTransactionStage, _transactionStatus, _pReloadNetworkDataContext));
		RETSUCCESS;
	}

	retT TransactionMgr::startTransactionStage( boost::shared_ptr<TransactionStage> _pTransactionStage, boost::weak_ptr<ReloadConnectionActor> _pReloadConnectionActor )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::shared_ptr<ReloadNetworkDataContext> pReloadNetworkDataContext(new ReloadNetworkDataContext());
		pReloadNetworkDataContext->pActor_ = _pReloadConnectionActor;
		R(processTransaction(_pTransactionStage, TransactionStatus(), pReloadOverlay, pReloadNetworkDataContext));
		RETSUCCESS;
	}

	retT TransactionMgr::getAndEraseTransaction( TransactionIdType _transactionId, Transaction& _transactionOut )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< TransactionIdType, Transaction >::iterator it = transactionMap_.find(_transactionId);
		if(it == transactionMap_.end())
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONID_NOT_EXIST);
		_transactionOut = it->second;
		RETSUCCESS;
	}

	retT TransactionMgr::startOperation( TransactionIdType _transactionId, TransactionOperationType _transactionOperation )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< TransactionIdType, Transaction >::iterator it = transactionMap_.find(_transactionId);
		if(it == transactionMap_.end())
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONID_NOT_EXIST);
		if(_transactionOperation != NULL)
		{
			it->second.transactionOperation_ = _transactionOperation;
			it->second.transactionStatus_.status_ = TransactionStatus_Operating;
			it->second.pTimerContext_.reset(new asyframe::HandlerContextBase());
			it->second.pTimerContext_->pTimerDuration_.reset(new boost::posix_time::seconds(TRANSMISSION_WAIT_SECONDS));
			R(pReloadOverlay->launchTimer(it->second.pTimerContext_, boost::bind(
				&TransactionMgr::onWaitOperationTimeout, shared_from_this(),
				_transactionId, it->second.pTimerContext_, _1, _2)));
			_transactionOperation();
		}
		else
		{
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONOPERATION_IS_NULL);
		}
		RETSUCCESS;
	}

	retT TransactionMgr::onWaitOperationTimeout( TransactionIdType _transactionId, boost::shared_ptr<asyframe::HandlerContextBase> _pTimerContext, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		R(handleTimeoutNoReset(_pTimerContext, e, _timeoutCount));
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< TransactionIdType, Transaction >::iterator it = transactionMap_.find(_transactionId);
		if(it == transactionMap_.end())
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONID_NOT_EXIST);
		if(!e)
		{
			if(_timeoutCount == 0)
			{
				it->second.transactionStatus_.status_ = TransactionStatus_Success;
			}
			else if(_timeoutCount <= RETRANSMISSION_TIMES)
			{
				R(pReloadOverlay->launchTimer(it->second.pTimerContext_, boost::bind(
					&TransactionMgr::onWaitOperationTimeout, shared_from_this(),
					_transactionId, it->second.pTimerContext_, _1, _2)));
				if(it->second.transactionOperation_ != NULL)
					it->second.transactionOperation_();
				else
					RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONOPERATION_IS_NULL);
			}
			else
			{
				it->second.transactionStatus_.status_ = TransactionStatus_OperationTimeout;
				R(switchToNextStage(_transactionId, it->second.pReloadNetworkDataContext_));
			}
		}
		else
			RET(INFO_COMMON_TIMER_FIRE_EXCEPTION);
		RETSUCCESS;
	}

	retT TransactionMgr::terminateTransaction( TransactionIdType _transactionId, TransactionStatus _transactionStatus /*= TransactionStatus_Success*/, boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext /*= boost::shared_ptr<ReloadNetworkDataContext>()*/ )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< TransactionIdType, Transaction >::iterator it = transactionMap_.find(_transactionId);
		if(it == transactionMap_.end())
			RET(INFO_PROTOCOL_RELOAD_TRANSACTIONMGR_TRANSACTIONID_NOT_EXIST);
		R(onWaitOperationTimeout(_transactionId, it->second.pTimerContext_, boost::system::error_code(), 0));
		it->second.transactionStatus_ = _transactionStatus;
		if(!_pReloadNetworkDataContext)
			_pReloadNetworkDataContext = it->second.pReloadNetworkDataContext_;
		R(switchToNextStage(_transactionId, _pReloadNetworkDataContext));
		RETSUCCESS;
	}

	///////////////////////////////////////////////////////////////////////////////////////////

	AttachReqTransactionStageParam::AttachReqTransactionStageParam()
	:pTransactionStage_(new TransactionStage()),pReloadConnectionActor_(),destinationType_(DestinationType_Node),
	connectFailedCallBackFun_(NULL)
	{

	}

	AttachReqTransactionStageParam::~AttachReqTransactionStageParam()
	{
		pTransactionStage_.reset();
		pReloadConnectionActor_.reset();
		destinationType_ = DestinationType_Node;
		connectFailedCallBackFun_ = NULL;
	}

	/////////////////////////////////////////////////////////////////////////////////////////////

	MessageCallBackMgr::MessageCallBackMgr( boost::weak_ptr<ReloadOverlay> _pReloadOverlay )
	:pReloadOverlay_(_pReloadOverlay),nodeMessageCallBackMap_()
	{

	}

	MessageCallBackMgr::~MessageCallBackMgr()
	{
		pReloadOverlay_.reset();
		nodeMessageCallBackMap_.clear();
	}

	retT MessageCallBackMgr::attachMessageCallBack( const NodeId& _nodeId, MessageCodeT _messageCode, const MessageCallBackFunType& _messageCallBackFun )
	{
		nodeMessageCallBackMap_[_nodeId][_messageCode].push_back(_messageCallBackFun);
		RETSUCCESS;
	}

	retT MessageCallBackMgr::invokeMessageCallBack( const NodeId& _nodeId, MessageCodeT _messageCode, boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		NodeMessageCallBackMap_::iterator itNodeId = nodeMessageCallBackMap_.find(_nodeId);
		if(itNodeId != nodeMessageCallBackMap_.end())
		{
			boost::unordered_map< MessageCodeT, std::vector< MessageCallBackFunType > >::iterator itMessageCode = itNodeId->second.find(_messageCode);
			if(itMessageCode != itNodeId->second.end())
			{
				if(boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock())
				{
					for(std::vector< MessageCallBackFunType >::iterator itMessageCallBackFun = itMessageCode->second.begin(); itMessageCallBackFun != itMessageCode->second.end(); ++itMessageCallBackFun)
					{
						R((*itMessageCallBackFun)(pReloadOverlay, _pReloadNetworkDataContext));
					}
					itMessageCode->second.clear();
				}
				else
					RET(ERROR_COMMON_SHARED_PTR_NULL);
			}
		}
		RETSUCCESS;
	}

	retT MessageCallBackMgr::erase( const NodeId& _nodeId )
	{
		nodeMessageCallBackMap_.erase(_nodeId);
		RETSUCCESS;
	}

	/////////////////////////////////////////////////////////////////////////////////////////////

	AttachStatus::AttachStatus()
	:attachStage_(AttachStage_Invalid),pReloadNetworkDataContext_(),pTimerContext_()
	{

	}

	AttachStatus::~AttachStatus()
	{
		attachStage_ = AttachStage_Invalid;
		pReloadNetworkDataContext_.reset();
		pTimerContext_.reset();
	}

	//////////////////////////////////////////////////////////////////////////////////////////////

	AttachMgr::AttachMgr( boost::weak_ptr<ReloadOverlay> _pReloadOverlay )
	:pReloadOverlay_(_pReloadOverlay),attachStatusMap_(),ipAddressPortIndexes_(),transactionsToStartAtConnectionEstablished_()
	{

	}

	AttachMgr::~AttachMgr()
	{
		attachStatusMap_.clear();
		ipAddressPortIndexes_.clear();
		transactionsToStartAtConnectionEstablished_.clear();
	}

	retT AttachMgr::checkAttachStatusWhileReqRecieved( boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay || !pReloadOverlay->pReloadConnectionActorMgr_ || !_pReloadNetworkDataContext || !_pReloadNetworkDataContext->pReloadPacket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_.size() == 0)
			RET(ERROR_COMMON_CONTAINER_SIZE_ERROR);
		boost::shared_ptr<AttachReqAns> pAttachReqAns = boost::dynamic_pointer_cast<AttachReqAns>(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
		if(!pAttachReqAns)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->getNodeId().isNULL())
			RETSUCCESS;
		NodeId nodeId = _pReloadNetworkDataContext->pReloadPacket_->getNodeId();
		if(nodeId == pReloadOverlay->nodeId_)
		{
			RET(INFO_PROTOCOL_RELOAD_MSG_SILENTLY_DROP);
		}
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(nodeId);
		if(it == attachStatusMap_.end())
		{
			for(std::size_t i = 0; i < pAttachReqAns->candidates_.array_.size(); ++i)
			{
				ReloadConnectionActorMgr::IpAddressPortIndexedMap_::iterator it = pReloadOverlay->pReloadConnectionActorMgr_->ipAddressPortIndexes_.find(pAttachReqAns->candidates_.array_[i].addrPort_);
				if(it != pReloadOverlay->pReloadConnectionActorMgr_->ipAddressPortIndexes_.end())
				{
					if(boost::shared_ptr<ReloadConnectionActor> pReloadConnectionActor = it->second.lock())
					{
						if(pReloadConnectionActor->getNodeId().isNULL())
							R(pReloadOverlay->setNodeIdToReloadConnectionActor(pReloadConnectionActor, nodeId));
					}
					R(pReloadOverlay->sendErrorResponse(_pReloadNetworkDataContext, Error_In_Progress));
					RET(INFO_PROTOCOL_RELOAD_ATTACH_IN_PROGRESS);
				}
			}
			R(addAttachStatus(_pReloadNetworkDataContext, AttachStage_ReqRecieved));
			RETSUCCESS;
		}
		switch(it->second.attachStage_)
		{
		case AttachStage_ReqRecieved:R(refreshAttachStatus(_pReloadNetworkDataContext));RETSUCCESS;
		case AttachStage_AnsRecieved:R(pReloadOverlay->sendErrorResponse(_pReloadNetworkDataContext, Error_In_Progress));RET(INFO_PROTOCOL_RELOAD_ATTACH_IN_PROGRESS);
		case AttachStage_ReqSended:
			{
				//compared as big-endian unsigned integers??????????
				if(pReloadOverlay->nodeId_ > nodeId)
				{
					R(pReloadOverlay->sendErrorResponse(_pReloadNetworkDataContext, Error_In_Progress));
					RET(INFO_PROTOCOL_RELOAD_ATTACH_IN_PROGRESS);
				}
				else
				{
					R(cancelAttachReq(nodeId));
					R(addAttachStatus(_pReloadNetworkDataContext, AttachStage_ReqRecieved));
					RETSUCCESS;
				}
			}
		default:break;
		}
		RET(ERROR_PROTOCOL_RELOAD_UNKNOWN_ATTACHSTATUS);
	}

	retT AttachMgr::addAttachStatus( boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext, int _attachStage )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay || !_pReloadNetworkDataContext || !_pReloadNetworkDataContext->pReloadPacket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_.size() == 0)
			RET(ERROR_COMMON_CONTAINER_SIZE_ERROR);
		boost::shared_ptr<AttachReqAns> pAttachReqAns = boost::dynamic_pointer_cast<AttachReqAns>(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
		if(!pAttachReqAns)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->getNodeId().isNULL())
			RETSUCCESS;
		NodeId nodeId = _pReloadNetworkDataContext->pReloadPacket_->getNodeId();
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(nodeId);
		if(it != attachStatusMap_.end())
			RET(ERROR_COMMON_CONTAINER_ELEMENT_ALREADY_EXIST);
		attachStatusMap_[nodeId].pReloadNetworkDataContext_ = _pReloadNetworkDataContext;
		attachStatusMap_[nodeId].attachStage_ = _attachStage;
		for(std::size_t i = 0; i < pAttachReqAns->candidates_.array_.size(); ++i)
		{
			ipAddressPortIndexes_[pAttachReqAns->candidates_.array_[i].addrPort_] = nodeId;
		}
		attachStatusMap_[nodeId].pTimerContext_.reset(new asyframe::HandlerContextBase());
		attachStatusMap_[nodeId].pTimerContext_->pTimerDuration_.reset(new boost::posix_time::seconds(MAXIMUM_REQUEST_LIFETIME_SECONDS));
		R(pReloadOverlay->launchTimer(attachStatusMap_[nodeId].pTimerContext_, boost::bind(
			&AttachMgr::onWaitConnectTimeout, shared_from_this(),
			nodeId, attachStatusMap_[nodeId].pTimerContext_, _1, _2)));
		RETSUCCESS;
	}

	retT AttachMgr::delAttachStatus( const NodeId& _nodeId )
	{
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(_nodeId);
		if(it == attachStatusMap_.end())
			RETSUCCESS;
		if(!it->second.pReloadNetworkDataContext_ || !it->second.pReloadNetworkDataContext_->pReloadPacket_)
			RETSUCCESS;
		boost::shared_ptr<AttachReqAns> pAttachReqAns = boost::dynamic_pointer_cast<AttachReqAns>(it->second.pReloadNetworkDataContext_->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
		if(!pAttachReqAns)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(it->second.pTimerContext_ && it->second.pTimerContext_->pHandler_ && it->second.pTimerContext_->timeoutCount_ == 0)
			R(onWaitConnectTimeout(_nodeId, it->second.pTimerContext_, boost::system::error_code(), 0));
		for(std::size_t i = 0; i < pAttachReqAns->candidates_.array_.size(); ++i)
		{
			ipAddressPortIndexes_.erase(pAttachReqAns->candidates_.array_[i].addrPort_);
		}
		attachStatusMap_.erase(it);
		RETSUCCESS;
	}

	retT AttachMgr::refreshAttachStatus( boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		if(!_pReloadNetworkDataContext || !_pReloadNetworkDataContext->pReloadPacket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->getNodeId().isNULL())
			RETSUCCESS;
		NodeId nodeId = _pReloadNetworkDataContext->pReloadPacket_->getNodeId();
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(nodeId);
		if(it == attachStatusMap_.end())
			RETSUCCESS;
		int attachStage = it->second.attachStage_;
		R(delAttachStatus(nodeId));
		R(addAttachStatus(_pReloadNetworkDataContext, attachStage));
		RETSUCCESS;
	}

	retT AttachMgr::cancelAttachReq( const NodeId& _nodeId )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(_nodeId);
		if(it == attachStatusMap_.end())
			RETSUCCESS;
		if(!it->second.pReloadNetworkDataContext_ || !it->second.pReloadNetworkDataContext_->pReloadPacket_)
			RETSUCCESS;
		Transaction transaction;
		R(pReloadOverlay->pTransactionMgr_->getAndEraseTransaction(it->second.pReloadNetworkDataContext_->pReloadPacket_->forwardingHeader_.transactionId_, transaction));
		attachStatusMap_.erase(it);
		RETSUCCESS;
	}

	retT AttachMgr::checkAttachStatusToSendAttachReq( const NodeId& _nodeId, boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext, boost::shared_ptr<AttachReqTransactionStageParam> _nextTransactionStageParam )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		ConnectionTable::NodeIdIndexedMap_::iterator it = pReloadOverlay->pConnectionTable_->nodeIdIndexes_.find(_nodeId);
		if(it != pReloadOverlay->pConnectionTable_->nodeIdIndexes_.end())
		{
			R(pReloadOverlay->startTransactionStage(_nextTransactionStageParam->pTransactionStage_, it->second));
			RET(INFO_PROTOCOL_RELOAD_ATTACH_IN_PROGRESS);
		}
		else
		{
			transactionsToStartAtConnectionEstablished_[_nodeId].push_back(_nextTransactionStageParam->pTransactionStage_);
			_nextTransactionStageParam->pTransactionStage_.reset(new TransactionStage());
			boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(_nodeId);
			if(it == attachStatusMap_.end())
			{
				attachStatusMap_[_nodeId].attachStage_ = AttachStage_ReqSended;
				attachStatusMap_[_nodeId].pReloadNetworkDataContext_ = _pReloadNetworkDataContext;
				attachStatusMap_[_nodeId].pTimerContext_.reset(new asyframe::HandlerContextBase());
				attachStatusMap_[_nodeId].pTimerContext_->pTimerDuration_.reset(new boost::posix_time::seconds(MAXIMUM_REQUEST_LIFETIME_SECONDS));
				R(pReloadOverlay->launchTimer(attachStatusMap_[_nodeId].pTimerContext_, boost::bind(
					&AttachMgr::onWaitConnectTimeout, shared_from_this(),
					_nodeId, attachStatusMap_[_nodeId].pTimerContext_, _1, _2)));
			}
			else
			{
				RET(INFO_PROTOCOL_RELOAD_ATTACH_IN_PROGRESS);
			}
		}
		RETSUCCESS;
	}

	retT AttachMgr::changeAttachStatusToAnsRecieved( boost::shared_ptr<ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay || !_pReloadNetworkDataContext || !_pReloadNetworkDataContext->pReloadPacket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_.size() == 0)
			RET(ERROR_COMMON_CONTAINER_SIZE_ERROR);
		boost::shared_ptr<AttachReqAns> pAttachReqAns = boost::dynamic_pointer_cast<AttachReqAns>(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
		if(!pAttachReqAns)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pReloadNetworkDataContext->pReloadPacket_->getNodeId().isNULL())
			RETSUCCESS;
		NodeId nodeId = _pReloadNetworkDataContext->pReloadPacket_->getNodeId();
		if(nodeId == pReloadOverlay->nodeId_)
		{
			RET(INFO_PROTOCOL_RELOAD_MSG_SILENTLY_DROP);
		}
		boost::unordered_map< NodeId, AttachStatus, boost::hash<NodeId> >::iterator it = attachStatusMap_.find(nodeId);
		if(it == attachStatusMap_.end())
		{
			R(addAttachStatus(_pReloadNetworkDataContext, AttachStage_AnsRecieved));
			RETSUCCESS;
		}
		else
		{
			if(it->second.attachStage_ == AttachStage_ReqSended)
			{
				attachStatusMap_.erase(it);
				R(addAttachStatus(_pReloadNetworkDataContext, AttachStage_AnsRecieved));
				RETSUCCESS;
			}
			else
				RET(INFO_PROTOCOL_RELOAD_UNEXPECTED_ATTACHSTATUS);
		}
		RETSUCCESS;
	}

	NodeId AttachMgr::getNodeIdFromIpAddressPort( const IpAddressPort& _ipAddressPort )
	{
		NodeId nodeId;
		boost::unordered_map< IpAddressPort, NodeId, boost::hash<IpAddressPort> >::iterator it = ipAddressPortIndexes_.find(_ipAddressPort);
		if(it != ipAddressPortIndexes_.end())
			nodeId = it->second;
		return nodeId;
	}

	retT AttachMgr::startTransactionsAtConnectionEstablished( const NodeId& _nodeId, boost::weak_ptr<ReloadConnectionActor> _pReloadConnectionActor )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::unordered_map< NodeId, std::vector< boost::shared_ptr<TransactionStage> >, boost::hash<NodeId> >::iterator it = transactionsToStartAtConnectionEstablished_.find(_nodeId);
		if(it != transactionsToStartAtConnectionEstablished_.end())
		{
			for(std::size_t i = 0; i < it->second.size(); ++i)
			{
				R(pReloadOverlay->pTransactionMgr_->startTransactionStage(it->second[i], _pReloadConnectionActor));
			}
			transactionsToStartAtConnectionEstablished_.erase(it);
		}
		RETSUCCESS;
	}

	retT AttachMgr::addTransactionToConnectionStartup( const NodeId& _nodeId, int _transactionStage, boost::shared_ptr<asyframe::WildcardElement> _param )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		boost::shared_ptr<TransactionStage> pTransactionStage(new TransactionStage());
		pTransactionStage->nextStage_ = _transactionStage;
		pTransactionStage->param_ = _param;
		ConnectionTable::NodeIdIndexedMap_::iterator it = pReloadOverlay->pConnectionTable_->nodeIdIndexes_.find(_nodeId);
		if(it == pReloadOverlay->pConnectionTable_->nodeIdIndexes_.end())
		{
			transactionsToStartAtConnectionEstablished_[_nodeId].push_back(pTransactionStage);
		}
		else
		{
			R(pReloadOverlay->startTransactionStage(pTransactionStage, it->second));
		}
		RETSUCCESS;
	}

	retT AttachMgr::eraseTransactionsInConnectionStartup( const NodeId& _nodeId )
	{
		transactionsToStartAtConnectionEstablished_.erase(_nodeId);
		RETSUCCESS;
	}

	retT AttachMgr::onWaitConnectTimeout( NodeId _nodeId, boost::shared_ptr<asyframe::HandlerContextBase> _pTimerContext, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		R(handleTimeoutNoReset(_pTimerContext, e, _timeoutCount));
		if(!e)
		{
			if(_timeoutCount > 0)
			{
				R(delAttachStatus(_nodeId));
			}
		}
		else
			RET(INFO_COMMON_TIMER_FIRE_EXCEPTION);
		RETSUCCESS;
	}

	///////////////////////////////////////////////////////////////////////////////////////////////

	DataStorageSetValue::DataStorageSetValue()
	:replicaNumber_(0)
	{

	}

	DataStorageSetValue::DataStorageSetValue( const DataStorageSetValue& _dataStorageSetValue )
		:resourceIdAsNodeIdDistance_(_dataStorageSetValue.resourceIdAsNodeIdDistance_),replicaNumber_(_dataStorageSetValue.replicaNumber_)
	{
		kindDataMap_.clear();
		for(boost::unordered_map<KindId, KindData>::const_iterator it = _dataStorageSetValue.kindDataMap_.begin(); it != _dataStorageSetValue.kindDataMap_.end(); ++it)
		{
			kindDataMap_.insert(*it);
		}
	}

	DataStorageSetValue::DataStorageSetValue( const ResourceId& _resourceId, const NodeId& _relativeNodeId )
	:replicaNumber_(0),resourceIdAsNodeIdDistance_(_resourceId, _relativeNodeId)
	{

	}

	DataStorageSetValue::~DataStorageSetValue()
	{
		replicaNumber_ = 0;
	}

	const DataStorageSetValue& DataStorageSetValue::operator=( const DataStorageSetValue& _dataStorageSetValue )
	{
		resourceIdAsNodeIdDistance_ = _dataStorageSetValue.resourceIdAsNodeIdDistance_;
		replicaNumber_ = _dataStorageSetValue.replicaNumber_;
		kindDataMap_.clear();
		for(boost::unordered_map<KindId, KindData>::const_iterator it = _dataStorageSetValue.kindDataMap_.begin(); it != _dataStorageSetValue.kindDataMap_.end(); ++it)
		{
			kindDataMap_.insert(*it);
		}
		return *this;
	}

	bool operator>( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ > _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	bool operator==( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ == _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	bool operator!=( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ != _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	bool operator<( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ < _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	bool operator>=( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ >= _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	bool operator<=( const DataStorageSetValue& _dataStorageSetValueLeft, const DataStorageSetValue& _dataStorageSetValueRight )
	{
		return _dataStorageSetValueLeft.resourceIdAsNodeIdDistance_ <= _dataStorageSetValueRight.resourceIdAsNodeIdDistance_;
	}

	///////////////////////////////////////////////////////////////////////////////////////////////

	DataStorageMgr::DataStorageMgr( boost::weak_ptr<ReloadOverlay> _pReloadOverlay )
	:pReloadOverlay_(_pReloadOverlay),dataStorageMap_()
	{

	}

	DataStorageMgr::~DataStorageMgr()
	{
		pReloadOverlay_.reset();
		dataStorageMap_.clear();
	}

	retT DataStorageMgr::getKindDataMap( const ResourceId& _resourceId, boost::unordered_map<KindId, KindData>& _kindDataMapOut )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		DataStorageMap_::iterator it = dataStorageMap_.find(ResourceIdAsNodeIdDistance(_resourceId, pReloadOverlay->nodeId_));
		if(it != dataStorageMap_.end())
		{
			_kindDataMapOut = it->kindDataMap_;
		}
		else
			RET(INFO_COMMON_ELEMENT_NOT_FOUND | NOLOG);
		RETSUCCESS;
	}

	retT DataStorageMgr::setKindDataMap( const ResourceId& _resourceId, const boost::unordered_map<KindId, KindData>& _kindDataMap, asyframe::uint8 _replicaNumber )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		DataStorageSetValue dataStorageSetValue;
		dataStorageSetValue.resourceIdAsNodeIdDistance_.resourceId_ = _resourceId;
		dataStorageSetValue.resourceIdAsNodeIdDistance_.relativeNodeId_ = pReloadOverlay->nodeId_;
		dataStorageSetValue.replicaNumber_ = _replicaNumber;
		dataStorageSetValue.kindDataMap_ = _kindDataMap;
		DataStorageMap_::iterator it = dataStorageMap_.find(dataStorageSetValue.resourceIdAsNodeIdDistance_);

		if(it != dataStorageMap_.end())
		{
			dataStorageMap_.replace(it, dataStorageSetValue);
		}
		else
		{
			dataStorageMap_.insert(dataStorageSetValue);
		}
		RETSUCCESS;
	}

	DataStorageMgr::DataStorageMap_::nth_index<1>::type& DataStorageMgr::getDataStorageMapIndexByResourceIdOrder()
	{
		return dataStorageMap_.get<1>();
	}

	//////////////////////////////////////////////////////////////////////////////////////////////////////////

	SendUpdateReqMgr::SendUpdateReqMgr(boost::weak_ptr<ReloadOverlay> _pReloadOverlay)
		:pReloadOverlay_(_pReloadOverlay)
	{

	}

	SendUpdateReqMgr::~SendUpdateReqMgr()
	{

	}

	retT SendUpdateReqMgr::sendUpdateReqToAllNodes()
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(!pReloadOverlay->pConnectionTable_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		for(UpdateReqMap_::iterator itUpdateReq = updateReqs.begin(); itUpdateReq != updateReqs.end(); ++itUpdateReq)
		{
			ConnectionTable::ConnectionMap_::iterator itConnection = pReloadOverlay->pConnectionTable_->connections_.find(itUpdateReq->first);
			if(itConnection != pReloadOverlay->pConnectionTable_->connections_.end())
			{
				for(boost::unordered_set<MessageCodeT>::iterator itMessageCode = itUpdateReq->second.begin(); itMessageCode != itUpdateReq->second.end(); ++itMessageCode)
				{
					boost::shared_ptr<ReloadNetworkDataContext> pReloadNetworkDataContext(new ReloadNetworkDataContext());
					pReloadNetworkDataContext->pActor_ = itConnection->second;
					R(pReloadOverlay->sendUpdateReq(pReloadNetworkDataContext, *itMessageCode));
				}
			}
		}
		updateReqs.clear();
		RETSUCCESS;
	}
} /*namespace reload*/
